package com.thiswhyme.mq.rocketmq.producer.producer;

import com.thiswhyme.mq.rocketmq.producer.consts.MqDestination;
import com.thiswhyme.mq.rocketmq.producer.dto.msg.TestMsgDto;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;

/**
 * Created by thiswhyme on 2020/6/7.
 */
@Component
public class ThwmMsgProducer {

    @Resource
    private RocketMQTemplate rocketMQTemplate;

    public void sendMessage() {
        sendCommonMsg(1);
        sendTransactionMsg(2);
        sendTagMsg(3);
        sendOneWayMsg(4);
        sendAsyncMsg(5);
        sendOrderlyMsg(6);
    }

    /**
     * 发送普通消息
     * @param dataId
     */
    private void sendCommonMsg(Integer dataId) {
        rocketMQTemplate.convertAndSend(MqDestination.TEST_COMMON_TOPIC,
                new TestMsgDto().setDataId(dataId).setContent("普通消息：" + dataId));
    }

    /**
     * 发送顺序消息
     */
    public void sendOrderlyMsg(Integer dataId){
        //arg：通过该参数指定发送到哪个队列，同队列消息保持顺序
        rocketMQTemplate.setMessageQueueSelector((List<MessageQueue> mqs, Message message, Object arg)->{
            int queueNum = Integer.valueOf(String.valueOf(arg)) % mqs.size();
            System.out.println("队列id："+queueNum+" 消息:"+new String(message.getBody()));
            return mqs.get(queueNum);
        });

        for (int i = dataId; i <= dataId + 10; i++) {
            rocketMQTemplate.syncSendOrderly(MqDestination.TEST_ORDERLY_TOPIC,
                    new TestMsgDto().setDataId(i).setContent("test" + i),
                    String.valueOf(1));
        }
    }

    /**
     * 发送事物消息
     * @param dataId
     */
    private TransactionSendResult sendTransactionMsg(Integer dataId) {
        TransactionSendResult result = rocketMQTemplate.sendMessageInTransaction(MqDestination.TEST_TRANSACTION_TOPIC,
                MessageBuilder.withPayload((new TestMsgDto().setDataId(dataId).setContent("事务消息：" + dataId))).build(),
                null);
        System.out.println("事务消息发送结果: " + result.getLocalTransactionState());
        return result;
    }

    /**
     * 发送带tag消息
     * @param dataId
     */
    private void sendTagMsg(Integer dataId) {
        rocketMQTemplate.convertAndSend(MqDestination.TEST_TAG_TOPIC + ":" +"tag2",
                new TestMsgDto().setDataId(dataId).setContent("tag2消息1：" + dataId));
        rocketMQTemplate.convertAndSend(MqDestination.TEST_TAG_TOPIC + ":" +"tag1",
                new TestMsgDto().setDataId(dataId).setContent("tag1消息1：" + dataId));
        rocketMQTemplate.convertAndSend(MqDestination.TEST_TAG_TOPIC + ":" +"tag2",
                new TestMsgDto().setDataId(dataId).setContent("tag2消息2：" + dataId));
    }

    /**
     * 发送带即发即失消息消息（不关心发送结果）
     * @param dataId
     */
    private void sendOneWayMsg(Integer dataId) {
        rocketMQTemplate.sendOneWay(MqDestination.TEST_COMMON_TOPIC + ":" +"tag",
                MessageBuilder.withPayload(new TestMsgDto().setDataId(dataId).setContent("即发即失消消息：" + dataId)).build());
    }

    /**
     * 发送异步消息
     * @param dataId
     */
    private void sendAsyncMsg(Integer dataId) {
        rocketMQTemplate.asyncSend(MqDestination.TEST_COMMON_TOPIC,
                MessageBuilder.withPayload(new TestMsgDto().setDataId(dataId).setContent("异步消息：" + dataId)).build(), new SendCallback() {
            @Override
            public void onSuccess(SendResult sendResult) {
                System.out.println("异步消息发送成功");
            }

            @Override
            public void onException(Throwable e) {
                System.out.println("异步消息发送失败");
            }
        });
    }
}
